package com.shujia.flink.core;

import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import java.util.HashMap;

/**
 * @author shujia
 */
public class Demo4Fraud {
    public static void main(String[] args) throws Exception {

        //创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        /*
         * 001,1687160239000,100
         * 001,1687160240000,0.2
         * 001,1687160241000,100
         * 001,1687160242000,600
         * 001,1687160243000,100
         * 001,1687160247000,0.6
         * 001,1687160248000,700
         * 001,1687160249000,100
         */
        //读取数据
        DataStreamSource<String> lines = env.socketTextStream("master", 8888);

        //解析数据
        DataStream<Transaction> transactions = lines.map(line -> {
            String[] split = line.split(",");
            String id = split[0];
            long tranTime = Long.parseLong(split[1]);
            Double tranMoney = Double.parseDouble(split[2]);
            return new Transaction(id, tranTime, tranMoney);
        });


        /*
         * 对于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息
         */

        /*
         * keyBy: 将相同的key分到同一个task中,同一个task中并不是只有一个key
         */
        //按照账号id分组
        KeyedStream<Transaction, String> tranKeyBy = transactions.keyBy(Transaction::getId);

        DataStream<Alert> alerts = tranKeyBy.process(new KeyedProcessFunction<String, Transaction, Alert>() {

            /*
            同一个task中的数据共享同一个成员变量
            private Boolean flag = false;
            private Double minTranMoney;
            */

            /**
             * 有状态计算: 每一次计算都会用到前一次计算的结果
             * 状态(前面的计算结果)
             */

            //使用hashmap为每一个key保存一个标记
            final HashMap<String, Boolean> flags = new HashMap<>(1000);

            //为每一个key保存小于1的金额
            final HashMap<String, Double> minTranMoneys = new HashMap<>(1000);

            final Double MAX_VALUE = 500.0;
            final Double MIN_VALUE = 1.0;

            /**
             * processElement每一条数据执行一次
             * @param tran 一行数据.
             * @param ctx 上下文对象,可以获取时间属性
             * @param out 用于将数据发生到下游
             */
            @Override
            public void processElement(Transaction tran,
                                       KeyedProcessFunction<String, Transaction, Alert>.Context ctx,
                                       Collector<Alert> out) {

                String id = tran.getId();
                long tranTime = tran.getTranTime();

                //从hashmap中取出标记
                Boolean flag = flags.getOrDefault(id, false);
                if (flag) {
                    //跟着一个大于 $500 的交易
                    if (tran.getTranMoney() > MAX_VALUE) {
                        System.out.println("跟着一个大于 $500 的交易");
                        //输出一个报警信息
                        System.out.println("输出一个报警信息");
                        //大于500的金额
                        Double maxTranMoney = tran.getTranMoney();
                        //小于1的金额
                        Double minTranMoney = minTranMoneys.get(id);
                        Alert alert = new Alert(id, tranTime, minTranMoney, maxTranMoney);
                        //发生数据到下游
                        out.collect(alert);
                    }
                    flags.put(id, false);
                }
                //如果出现小于 $1 美元的交易
                if (tran.getTranMoney() < MIN_VALUE) {
                    System.out.println("出现小于 $1 美元的交易");
                    //flag = true;
                    flags.put(id, true);
                    //将小于1的交易保持起来
                    minTranMoneys.put(id, tran.getTranMoney());
                }
            }
        });

        alerts.print();

        env.execute();


    }

    @Data
    @AllArgsConstructor
    static class Transaction {
        private String id;
        private long tranTime;
        private Double tranMoney;
    }

    @Data
    @AllArgsConstructor
    static class Alert {
        private String id;
        private long tranTime;
        private Double minTranMoney;
        private Double maxTranMoney;
    }
}
